/*
 * BaseRemoteInvocationDispatcher.java    0.0.1    Nov 4, 2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.net.ril.impl;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;

import javax.management.JMException;
import javax.management.MBeanServer;
import javax.management.ObjectName;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.commons.timer.MonitorTimePerformances;
import org.mentalsmash.tazio.commons.timer.TimingsLog;
import org.mentalsmash.tazio.net.ril.DispatchingLayerProxy;
import org.mentalsmash.tazio.net.ril.RemoteInvocationDispatcher;
import org.mentalsmash.tazio.net.ril.RemoteInvocationMessageHandler;
import org.mentalsmash.tazio.net.ril.RemoteMethod;
import org.mentalsmash.tazio.net.ril.RemoteObject;
import org.mentalsmash.tazio.net.ril.RemoteObjectProxy;
import org.mentalsmash.tazio.net.ril.RemoteObjectsRegistry;
import org.mentalsmash.tazio.net.ril.RemoteMethod.InvocationSemantics;
import org.mentalsmash.tazio.net.ril.exceptions.FailedMessageDeliveryException;
import org.mentalsmash.tazio.net.ril.exceptions.NoResponseReceivedException;
import org.mentalsmash.tazio.net.ril.exceptions.NoSuchRemoteMethodException;
import org.mentalsmash.tazio.net.ril.exceptions.NoSuchRemoteObjectException;
import org.mentalsmash.tazio.net.ril.exceptions.NotSerializableArgumentException;
import org.mentalsmash.tazio.net.ril.exceptions.NotSerializableOutcomeException;
import org.mentalsmash.tazio.net.ril.exceptions.UnavailableRemoteObjectException;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RILResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RemoteObjectRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RemoteObjectResponseMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.name.Named;

/**
 * This class provides a basic implementation of the
 * {@link RemoteInvocationDispatcher} interface. It has an implementation of
 * both methods declared by this interface and allows subclasses to customize
 * the way messages are sent to other hosts by providing a custom instance of
 * {@link DispatchingLayerProxy}. method.
 * 
 * @version 0.0.1 Nov 4, 2009
 * @author Andrea Sorbini <as@mentalsmash.org>
 * 
 */
@Concurrency(concurrentAccess = true, specification = "public methods available to client for requests"
	+ "sending have no concurrency concern as they merely take care of building an appropriate request"
	+ "message and send it/receive its response by means of a DispatchingLayerProxy. anyway access to these methods "
	+ "might (and most probably will) happen concurrently and since they all update the monitored class status "
	+ "(i.e. the one exposed by the MBean interface), private methods that access the MBean-related fields synchronize "
	+ "on two lock Objects, depending on the \"semantics\" of the information to update (i.e. if the "
	+ "field to update is related to RemoteObjects requests the remoteObjectsLock is used, "
	+ "if it concerns Method Invocations requests synchronization occurs on the invocationsLock")
class BaseRemoteInvocationDispatcher implements RemoteInvocationDispatcher,
	BaseRemoteInvocationDispatcherMBean {

    private static final Logger log = LoggerFactory
	    .getLogger(BaseRemoteInvocationDispatcher.class);

    // protected final RemoteInvoker invoker;

    private final ProxyFixer fixer;
    private final DispatchingLayerProxy dispatchingLayer;

    private final String name;

    // MBean-related fields

    private final Object invocationsLock = new Object();
    private final Object remoteObjectsLock = new Object();

    private int requestedRemoteObjects;
    private int receivedRemoteObjects;
    private int failedRemoteObjects;

    private int requestedMethodInvocations;
    private int executedMethodInvocations;
    private int failedMethodInvocations;

    private int requestedGroupMethodInvocations;
    private int completedGroupMethodInvocations;
    private int failedGroupMethodInvocations;

    private final TimingsLog tlog;

    public static final String TLOG_KEY_REMOTEINVOC = BaseRemoteInvocationDispatcher.class
	    .getName()
	    + "#" + "invokeRemoteMethod";

    private ObjectName mbeanName;

    public ObjectName getMBeanName() {
	return this.mbeanName;
    }

    /**
     * Creates a new BaseRemoteInvocationDispatcher connected to the supplied
     * {@link RemoteObjectsRegistry} and using the supplied
     * {@link RemoteInvocationMessageHandler} to receive responses from other
     * hosts.
     * 
     * @param invoker
     */
    @Inject
    BaseRemoteInvocationDispatcher(ProxyFixer fixer,
	    DispatchingLayerProxy dispatchingLayer,
	    @Named(RILBaseImplModule.DISPATCHER_NAME) String name,
	    TimingsLog tlog) {
	// this.invoker = invoker;
	// this.invoker.setDispatcher(this);
	this.fixer = fixer;
	this.dispatchingLayer = dispatchingLayer;
	this.name = name;
	this.tlog = tlog;
    }

    /**
     * @see RemoteInvocationDispatcher#getRemoteObject(OID)
     */
    @Override
    public RemoteObject getRemoteObject(OID oid)
	    throws NoSuchRemoteObjectException,
	    UnavailableRemoteObjectException {

	RemoteObjectRequestMessage msg = new RemoteObjectRequestMessage(oid);

	RILResponseMessage response = null;

	try {
	    log.debug("Sending request for remote object " + oid);

	    // DispatchingLayerProxy dispatchingLayer = this.invoker
	    // .getDispatchingLayerProxy();

	    this.requestedRemoteObject();

	    response = this.dispatchingLayer.sendMessage(msg);

	} catch (FailedMessageDeliveryException ex) {

	    this.failedRemoteObject();

	    log.error("Cannot deliver message to request object " + oid);
	    throw new UnavailableRemoteObjectException(oid);
	} catch (NoResponseReceivedException e) {
	    this.failedRemoteObject();

	    log.error("No response received to retrieve object " + oid);
	    throw new UnavailableRemoteObjectException(oid);
	}

	RemoteObjectResponseMessage resp = (RemoteObjectResponseMessage) response;

	if (resp.getRemoteObject() == null) {

	    this.failedRemoteObject();

	    log.warn("Remote object " + oid + " doesn't exist");
	    throw new NoSuchRemoteObjectException(oid);
	} else {
	    RemoteObject received = resp.getRemoteObject();

	    // RemoteObjectsRegistry registry = this.invoker.getRegistry();

	    RemoteObjectProxy proxy = received.getProxy();

	    // proxy.setRegistry(registry);
	    // proxy.setDispatcher(this);

	    this.fixer.fix(proxy);

	    this.receivedRemoteObject();

	    log.debug("Remote object " + oid + " received");
	    return received;
	}
    }

    /**
     * @see RemoteInvocationDispatcher#invokeRemoteMethod(OID, RemoteMethod,
     *      Object[])
     */
    @MonitorTimePerformances
    @Override
    public Object invokeRemoteMethod(OID oid, RemoteMethod method, Object[] args)
	    throws NoSuchRemoteObjectException, NoSuchRemoteMethodException,
	    NotSerializableArgumentException, NotSerializableOutcomeException {

	if (method != null) {
	    if (!method.checkArgs(args))
		throw new IllegalArgumentException(
			"Illegal arguments for method " + method);
	} else
	    throw new IllegalArgumentException("null method to invoke");

	if (oid == null) {
	    throw new IllegalArgumentException(
		    "null oid to invoke the method on");
	}

	ArrayList<Serializable> arguments = new ArrayList<Serializable>();

	// if (!method.isByCopy()) {
	// args = this.checkArgumentsForRemoteObjects(args);
	// }

	for (Object arg : args) {
	    if (!(arg instanceof Serializable)) {
		throw new NotSerializableArgumentException(arg);
	    } else {
		arguments.add((Serializable) arg);
	    }
	}

	MethodInvocationRequestMessage msg = new MethodInvocationRequestMessage(
		oid, method, arguments.toArray(new Serializable[0]));

	StringBuilder argsStr = new StringBuilder();
	Iterator<Serializable> argsIter = arguments.iterator();
	while (argsIter.hasNext()) {
	    argsStr.append(argsIter.next());
	    if (argsIter.hasNext()) {
		argsStr.append(", ");
	    }
	}

	if (method.getDeclaringClass().equals(Object.class)) {
	    log.trace("Requesting remote invocation : on {} call {}",
		    new Object[] { oid, method });
	} else {
	    log.debug("Requesting remote invocation on {} call {} args {}",
		    new Object[] { oid, method, argsStr.toString() });
	}

	RILResponseMessage response = null;

	// DispatchingLayerProxy dispatchingLayer = this.invoker
	// .getDispatchingLayerProxy();

	boolean onGroup = method.getSemantics().equals(
		InvocationSemantics.OnGroup);

	try {

	    this.requestedMethodInvocation(onGroup);

	    switch (method.getSemantics()) {
	    case Default:
		response = dispatchingLayer.sendMessage(msg);
		break;

	    case AtMostOnce:
		response = dispatchingLayer.sendMethodInvocationAtMostOnce(msg);
		break;

	    case OnGroup:
		response = dispatchingLayer.sendMethodInvocationToGroup(msg);
		break;
	    }

	} catch (FailedMessageDeliveryException ex) {

	    this.failedMethodInvocation(onGroup);

	    log.error("Cannot deliver message to invoke " + method + " on "
		    + oid);
	    return new UnavailableRemoteObjectException(oid);

	} catch (NoResponseReceivedException e) {
	    this.failedMethodInvocation(onGroup);

	    log.error("No response received for invocation " + method + " on "
		    + oid);
	    return new UnavailableRemoteObjectException(oid);
	}

	MethodInvocationResponseMessage outcMsg = (MethodInvocationResponseMessage) response;

	if (outcMsg.raisedException()) {
	    Exception e = (Exception) outcMsg.getOutcome();

	    if (e instanceof NoSuchRemoteObjectException) {

		this.failedMethodInvocation(onGroup);

		log.warn("Seems like we probably requested "
			+ "method's invocation to the wrong remote host...");
		throw (NoSuchRemoteObjectException) e;
	    } else if (e instanceof NoSuchRemoteMethodException) {

		this.failedMethodInvocation(onGroup);

		log.warn("Remote method " + method
			+ " not found on remote instance " + oid);
		throw (NoSuchRemoteMethodException) e;
	    } else if (e instanceof NotSerializableOutcomeException) {

		this.failedMethodInvocation(onGroup);

		log.warn("Remote method returned a not serializable outcome");
		throw (NotSerializableOutcomeException) e;
	    } else {
		log.warn("Remote method invocation raised an exception : "
			+ e.getClass().getCanonicalName() + " (message: "
			+ e.getMessage() + ")");
		return e;
	    }

	} else {
	    Object outcome = null;

	    if (method.getReturnType().equals(Void.TYPE)) {
		outcome = "void";
	    } else {
		outcome = outcMsg.getOutcome();
	    }

	    if (method.getDeclaringClass().equals(Object.class)) {
		log.trace("Invocation's outcome : " + outcome);
	    } else {
		log.debug("Invocation's outcome : " + outcome);
	    }

	    this.executedMethodInvocation(onGroup);

	    return outcMsg.getOutcome();
	}

    }

    // protected Object[] checkArgumentsForRemoteObjects(Object[] args) {
    // if (args != null) {
    // RemoteObjectsRegistry registry = this.invoker.getRegistry();
    //
    // ArrayList<Object> arguments = new ArrayList<Object>();
    //
    // for (Object o : args) {
    //
    // OID argOID = registry.getOID(o);
    //
    // if (argOID != null) {
    //
    // try {
    // RemoteObject remoteObj = registry
    // .getRemoteObject(argOID);
    // arguments.add(remoteObj);
    // } catch (NoSuchRemoteObjectException ex) {
    // // should never happen, since we've checked object
    // // presence with getOID()
    // }
    //
    // } else {
    // arguments.add(o);
    // }
    //
    // }
    //
    // return arguments.toArray();
    //
    // } else {
    // return new Object[0];
    // }
    // }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getExecutedRemoteInvocations()
     */
    @Override
    public int getRemoteInvocationsCompleted() {
	synchronized (this.invocationsLock) {
	    return this.executedMethodInvocations;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getFailedRemoteInvocations()
     */
    @Override
    public int getRemoteInvocationsFailed() {
	synchronized (this.invocationsLock) {
	    return this.failedMethodInvocations;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getInvalidRemoteObjects()
     */
    @Override
    public int getRemoteObjectsInvalid() {
	synchronized (this.remoteObjectsLock) {
	    return this.failedRemoteObjects;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getReceivedRemoteObjects()
     */
    @Override
    public int getRemoteObjectsReceived() {
	synchronized (this.remoteObjectsLock) {
	    return this.receivedRemoteObjects;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getRequestedRemoteInvocations()
     */
    @Override
    public int getRemoteInvocationsRequested() {
	synchronized (this.invocationsLock) {
	    return this.requestedMethodInvocations;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.RemoteInvocationDispatcherMBean#
     * getRequestedRemoteObjects()
     */
    @Override
    public int getRemoteObjectsRequested() {
	synchronized (this.remoteObjectsLock) {
	    return this.requestedRemoteObjects;
	}
    }

    private void requestedMethodInvocation(boolean onGroup) {
	synchronized (this.invocationsLock) {
	    if (onGroup) {
		this.requestedGroupMethodInvocations++;
	    } else {
		this.requestedMethodInvocations++;
	    }
	}
    }

    private void failedMethodInvocation(boolean onGroup) {
	synchronized (this.invocationsLock) {
	    if (onGroup) {
		this.failedGroupMethodInvocations++;
	    } else {
		this.failedMethodInvocations++;
	    }
	}
    }

    private void executedMethodInvocation(boolean onGroup) {
	synchronized (this.invocationsLock) {
	    if (onGroup) {
		this.completedGroupMethodInvocations++;
	    } else {
		this.executedMethodInvocations++;
	    }
	}
    }

    private void requestedRemoteObject() {
	synchronized (this.remoteObjectsLock) {
	    this.requestedRemoteObjects++;
	}
    }

    private void failedRemoteObject() {
	synchronized (this.remoteObjectsLock) {
	    this.failedRemoteObjects++;
	}
    }

    private void receivedRemoteObject() {
	synchronized (this.remoteObjectsLock) {
	    this.receivedRemoteObjects++;
	}
    }

    /**
     * This method is used to inject an MBeanServer instance that will be used
     * to register this MBean
     */
    @Inject
    public void register(MBeanServer server) {
	try {
	    this.mbeanName = new ObjectName(
		    "org.mentalsmash.tazio.net.ril:type="
			    + "RemoteInvocationDispatcher");
	    server.registerMBean(this, this.mbeanName);
	} catch (JMException e) {
	    System.err.println("exception: " + e.getClass().getCanonicalName()
		    + " " + e.getMessage());
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.ril.impl.BaseRemoteInvocationDispatcherMBean
     * #getGroupRemoteInvocationsCompleted()
     */
    @Override
    public int getGroupRemoteInvocationsCompleted() {
	synchronized (invocationsLock) {
	    return this.completedGroupMethodInvocations;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.ril.impl.BaseRemoteInvocationDispatcherMBean
     * #getGroupRemoteInvocationsFailed()
     */
    @Override
    public int getGroupRemoteInvocationsFailed() {
	synchronized (invocationsLock) {
	    return this.failedGroupMethodInvocations;
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.ril.impl.BaseRemoteInvocationDispatcherMBean
     * #getGroupRemoteInvocationsRequested()
     */
    @Override
    public int getGroupRemoteInvocationsRequested() {
	synchronized (invocationsLock) {
	    return this.requestedGroupMethodInvocations;
	}
    }

    @Override
    public double getAvgRemoteInvocationTime() {
	return this.tlog.getAvgMeasure(TLOG_KEY_REMOTEINVOC);
    }

    @Override
    public long getMaxRemoteInvocationTime() {
	return this.tlog.getMaxMeasure(TLOG_KEY_REMOTEINVOC);
    }

    @Override
    public long getMinRemoteInvocationTime() {
	return this.tlog.getMinMeasure(TLOG_KEY_REMOTEINVOC);
    }

}
